package com.hebeiwangxiaozhang.admin.service;

import com.alibaba.fastjson.JSONArray;
import com.hebeiwangxiaozhang.admin.feign.ProductServiceClient;
import com.hebeiwangxiaozhang.admin.pojo.*;
import com.hebeiwangxiaozhang.admin.repo.*;
import com.hebeiwangxiaozhang.admin.constants.GlobalConstants;
import com.hebeiwangxiaozhang.admin.utils.RedisCommonProcessor;
import com.hebeiwangxiaozhang.pojo.CommodityAttributes;
import com.hebeiwangxiaozhang.pojo.CommodityEsBodyDeprecated;
import com.hebeiwangxiaozhang.response.CommonResponse;
import com.hebeiwangxiaozhang.response.ResponseCode;
import com.hebeiwangxiaozhang.response.ResponseUtils;
import com.hebeiwangxiaozhang.utils.JsonUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.transaction.Transactional;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Service
public class CommodityService {

    @Autowired
    private CommoditySkuRepository commoditySkuRepository;

    @Autowired
    private CommoditySpuRepository commoditySpuRepository;

    @Autowired
    private CommoditySpuDetailRepository commoditySpuDetailRepository;

    @Autowired
    private CommodityBrandRepository commodityBrandRepository;

    @Autowired
    private CommodityCategoryRepository commodityCategoryRepository;

    @Autowired
    private CommodityAttrRepository commodityAttrRepository;

    @Autowired
    private CommodityAttrGroupRepository commodityAttrGroupRepository;

    @Autowired
    private RedisCommonProcessor redisCommonProcessor;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ProductServiceClient productServiceClient;

    @Autowired
    private CommoditySkuUpErrorRepository commoditySkuUpErrorRepository;

    @Transactional
    public void createCommodity(CommodityParam commodityParam) {
        CommoditySpu commoditySpu = CommoditySpu.builder()
                .name(commodityParam.getName())
                .isSale(GlobalConstants.STATUS_FALSE)
                .brandId(commodityParam.getBrandId())
                .categoryId(commodityParam.getCategoryId())
                .createTime(new Date())
                .updateTime(new Date())
                .build();
        commoditySpuRepository.save(commoditySpu);

        CommoditySpuDetail commoditySpuDetail = commodityParam.getSpuDetail();
        commoditySpuDetail.setSpuId(commoditySpu.getId());
        commoditySpuDetail.setCreateTime(new Date());
        commoditySpuDetail.setUpdateTime(new Date());

        commoditySpuDetailRepository.save(commoditySpuDetail);

        Map<String, List<String>> spcAttr = JsonUtils.readValue(commoditySpuDetail.getSpecificAttr(), Map.class);
        AtomicInteger pos = new AtomicInteger(0);
        List<CommoditySku> commoditySkuList = commodityParam.getSkus().stream()
                .map(c -> {
                    c.setSpuId(commoditySpu.getId());
                    c.setCreateTime(new Date());
                    c.setUpdateTime(new Date());
                    String[] indexes = c.getIndexes().split("_");
                    Map<String, String> ownAttrMap = new HashMap<>();
                    spcAttr.keySet().stream().forEach(key -> {
                        ownAttrMap.put(key, spcAttr.get(key).get(Integer.valueOf(indexes[pos.getAndIncrement()])));
                    });
                    pos.set(0);
                    c.setOwnAttr(JsonUtils.toJSONString(ownAttrMap));
                    c.setIsValid(GlobalConstants.STATUS_TRUE);
                    return c;
                }).collect(Collectors.toList());

        commoditySkuRepository.saveAll(commoditySkuList);
    }

    @Transactional
    public void updateCommodity(CommodityParam commodityParam) {
        Long spuId = commodityParam.getSpuDetail().getSpuId();

        Optional<CommoditySpu> spuOptional = commoditySpuRepository.findById(spuId);
        if (spuOptional.isPresent()) {
            CommoditySpu commoditySpu = spuOptional.get();

            CommoditySpu spu = CommoditySpu.builder()
                    .id(commoditySpu.getId())
                    .name(commodityParam.getName())
                    .isSale(GlobalConstants.STATUS_FALSE)
                    .brandId(commodityParam.getBrandId())
                    .categoryId(commodityParam.getCategoryId())
                    .createTime(commoditySpu.getCreateTime())
                    .updateTime(new Date())
                    .build();
            commoditySpuRepository.save(spu);
        }

        Optional<CommoditySpuDetail> spuDetailOptional = commoditySpuDetailRepository.findById(spuId);
        if (spuDetailOptional.isPresent()) {
            CommoditySpuDetail commoditySpuDetail = spuDetailOptional.get();

            CommoditySpuDetail spuDetail = commodityParam.getSpuDetail();
            spuDetail.setUpdateTime(new Date());
            spuDetail.setCreateTime(commoditySpuDetail.getCreateTime());
            spuDetail.setSpuId(commoditySpuDetail.getSpuId());
            commoditySpuDetailRepository.save(spuDetail);
        }


        List<CommoditySku> commoditySkuList = commoditySkuRepository.getBySpuId(spuId);
        commoditySkuRepository.deleteAll(commoditySkuList);

        Map<String, List<String>> spcAttr = JsonUtils.readValue(commodityParam.getSpuDetail().getSpecificAttr(), Map.class);
        AtomicInteger pos = new AtomicInteger(0);
        List<CommoditySku> skuList = commodityParam.getSkus().stream()
                .map(c -> {
                    c.setSpuId(spuId);
                    c.setCreateTime(new Date());
                    c.setUpdateTime(new Date());
                    String[] indexes = c.getIndexes().split("_");
                    Map<String, String> ownAttrMap = new HashMap<>();
                    spcAttr.keySet().stream().forEach(key -> {
                        ownAttrMap.put(key, spcAttr.get(key).get(Integer.valueOf(indexes[pos.getAndIncrement()])));
                    });
                    pos.set(0);
                    c.setOwnAttr(JsonUtils.toJSONString(ownAttrMap));
                    c.setIsValid(GlobalConstants.STATUS_TRUE);
                    return c;
                }).collect(Collectors.toList());

        commoditySkuRepository.saveAll(skuList);

    }

    public List<CommoditySku> getSkuBySpuId(Long spuId) {
        List<CommoditySku> commoditySkuList = commoditySkuRepository.getBySpuId(spuId);
        return commoditySkuList;
    }

    public CommoditySpuDetail getSpuDetailBySpuId(Long spuId) {
        Optional<CommoditySpuDetail> commoditySpuDetailOptional = commoditySpuDetailRepository.findById(spuId);
        if (commoditySpuDetailOptional.isPresent()) {
            return commoditySpuDetailOptional.get();
        }
        return null;
    }

    // 这个 esBody 我们最终传输给 commodity-service，这个commodity-service和ES 直接进行交互
    //不着急，首先进行esBodies的拼装，并且在这个过程中进行调优优化。
    @Deprecated
    public void onSale(Long spuId) {
        Optional<CommoditySpu> spu = commoditySpuRepository.findById(spuId);
        Assert.isTrue(spu.isPresent(), "SPU should not be null!");
        Optional<CommoditySpuDetail> spuDetail = commoditySpuDetailRepository.findById(spuId);
        Assert.isTrue(spuDetail.isPresent(), "spuDetail should not be null!");
        List<CommoditySku> skus = commoditySkuRepository.getBySpuId(spuId);
        Assert.notEmpty(skus, "skus should not be null!");
        Optional<CommodityBrand> brand = commodityBrandRepository.findById(spu.get().getBrandId());
        Assert.isTrue(brand.isPresent(), "brand should not be null!");
        Optional<CommodityCategory> category = commodityCategoryRepository.findById(spu.get().getCategoryId());
        Assert.isTrue(category.isPresent(), "category should not be null!");

        //公共的 attrs 查一次就行
        Map<String, String> genericAttrs = JsonUtils.readValue(spuDetail.get().getGenericAttr(), Map.class);
        List<CommodityAttributes> attrs = new ArrayList<>();
        genericAttrs.keySet().stream().forEach(key -> {
            Optional<CommodityAttr> attr = commodityAttrRepository.findById(Long.parseLong(key));
            Optional<CommodityAttrGroup> attrGroup = commodityAttrGroupRepository.findById(attr.get().getGroupId());
            CommodityAttributes commodityAttributes = CommodityAttributes.builder()
                    .attrId(attr.get().getId())
                    .attrName(attr.get().getName())
                    .attrGroupId(attrGroup.get().getId())
                    .attrGroupName(attrGroup.get().getName())
                    .attrValue(genericAttrs.get(key))
                    .build();
            attrs.add(commodityAttributes);
        });
        List<CommodityEsBodyDeprecated> esBodies = new ArrayList<>();
        skus.forEach(sku -> {
            // 特殊的 attrs 需要根据 sku 具体的值
            Map<String, String> ownAttr = JsonUtils.readValue(sku.getOwnAttr(), Map.class);
            List<CommodityAttributes> skuAttrs = new ArrayList<>(attrs);
            ownAttr.keySet().stream().forEach(key -> {
                Optional<CommodityAttr> attr = commodityAttrRepository.findById(Long.parseLong(key));
                Optional<CommodityAttrGroup> attrGroup = commodityAttrGroupRepository.findById(attr.get().getGroupId());
                CommodityAttributes commodityAttributes = CommodityAttributes.builder()
                        .attrId(attr.get().getId())
                        .attrName(attr.get().getName())
                        .attrGroupId(attrGroup.get().getId())
                        .attrGroupName(attrGroup.get().getName())
                        .attrValue(ownAttr.get(key))
                        .build();
                skuAttrs.add(commodityAttributes);
            });

            CommodityEsBodyDeprecated esBody = CommodityEsBodyDeprecated
                    .builder()
                    .brandId(spu.get().getBrandId())
                    .categoryId(spu.get().getCategoryId())
                    .spuId(spuId)
                    .skuId(sku.getId())
                    .price(sku.getPrice())
                    .stock(sku.getStock())
                    .brandName(brand.get().getName())
                    .categoryName(category.get().getName())
                    .image("TO DO")
                    .skuTitle(sku.getTitle())
                    .packingList(spuDetail.get().getPackingList())
                    .customerService(spuDetail.get().getCustomerService())
                    .attrs(skuAttrs)
                    .build();
            esBodies.add(esBody);
        });
        System.out.println(JsonUtils.toJSONString(esBodies));
    }

    // 优化点：attrs 我们需要把它放到缓存中，并且是永不过期的。对于第一次调用的时候，查DB，但是第二次直接从缓存中获取
    @Deprecated
    public void onSale1(Long spuId) {
        Optional<CommoditySpu> spu = commoditySpuRepository.findById(spuId);
        Assert.isTrue(spu.isPresent(), "SPU should not be null!");
        Optional<CommoditySpuDetail> spuDetail = commoditySpuDetailRepository.findById(spuId);
        Assert.isTrue(spuDetail.isPresent(), "spuDetail should not be null!");
        List<CommoditySku> skus = commoditySkuRepository.getBySpuId(spuId);
        Assert.notEmpty(skus, "skus should not be null!");
        Optional<CommodityBrand> brand = commodityBrandRepository.findById(spu.get().getBrandId());
        Assert.isTrue(brand.isPresent(), "brand should not be null!");
        Optional<CommodityCategory> category = commodityCategoryRepository.findById(spu.get().getCategoryId());
        Assert.isTrue(category.isPresent(), "category should not be null!");

        //公共的 attrs 查一次就行
        Map<String, String> genericAttrs = JsonUtils.readValue(spuDetail.get().getGenericAttr(), Map.class);
        Map<String, CommodityAttributes> attrsCache = getAttrsCacheRedisson();
        List<CommodityAttributes> attrs = new ArrayList<>();
        genericAttrs.keySet().stream().forEach(key -> {
            CommodityAttributes commodityAttributes = attrsCache.get(key);
            commodityAttributes.setAttrId(Long.parseLong(key));
            commodityAttributes.setAttrValue(genericAttrs.get(key));
            attrs.add(commodityAttributes);
        });
        List<CommodityEsBodyDeprecated> esBodies = new ArrayList<>();
        skus.forEach(sku -> {
            // 特殊的 attrs 需要根据 sku 具体的值
            Map<String, String> ownAttr = JsonUtils.readValue(sku.getOwnAttr(), Map.class);
            List<CommodityAttributes> skuAttrs = new ArrayList<>(attrs);
            ownAttr.keySet().stream().forEach(key -> {
                CommodityAttributes commodityAttributes = attrsCache.get(key);
                commodityAttributes.setAttrId(Long.parseLong(key));
                commodityAttributes.setAttrValue(ownAttr.get(key));
                skuAttrs.add(commodityAttributes);
            });
            CommodityEsBodyDeprecated esBody = CommodityEsBodyDeprecated
                    .builder()
                    .brandId(spu.get().getBrandId())
                    .categoryId(spu.get().getCategoryId())
                    .spuId(spuId)
                    .skuId(sku.getId())
                    .price(sku.getPrice())
                    .stock(sku.getStock())
                    .brandName(brand.get().getName())
                    .categoryName(category.get().getName())
                    .image("TO DO")
                    .skuTitle(sku.getTitle())
                    .packingList(spuDetail.get().getPackingList())
                    .customerService(spuDetail.get().getCustomerService())
                    .attrs(skuAttrs)
                    .build();
            esBodies.add(esBody);
        });
        System.out.println(JsonUtils.toJSONString(esBodies));
    }

    //这个方法有问题吗？引出新的技术点
    //如果 缓存 中的数据没有了。（新增 属性或者属性组，为了保证双写一致性，是不是我们都是会删除 redis中的缓存啊? 还有一些未知的
    // 数据异常丢失） 如果这时候，有很多客户端同事访问商品上架功能，那么这时候if(cache != null) 都不会进去，都会直接查询DB。
    // 缓存击穿问题
    @Deprecated
    private Map<String, CommodityAttributes> getAttrsCache() {
        Object cache = redisCommonProcessor.get("attrsCache");
        if(cache != null) {
            return (Map<String, CommodityAttributes>) cache;
        }
        Map<String, CommodityAttributes> attrsCache = new HashMap<>();
        List<CommodityAttr> attrs = commodityAttrRepository.findAll();
        attrs.forEach(attr -> {
            Optional<CommodityAttrGroup> attrGroup = commodityAttrGroupRepository.findById(attr.getGroupId());
            CommodityAttributes commodityAttributes = CommodityAttributes.builder()
                    .attrName(attr.getName())
                    .attrGroupName(attrGroup.get().getName())
                    .attrGroupId(attrGroup.get().getId())
                    .build();
            attrsCache.put(attr.getId().toString(), commodityAttributes);
        });

        redisCommonProcessor.set("attrsCache", attrsCache);
        return attrsCache;
    }

    //第一种方式，万恶的 sync、lock
    //db同时最多有 分布式节点个数 个访问同时打向DB。 但是这个东西，在正常情况下，依然会导致多线程的锁的争抢问题，性能极低。
    //小伙伴提出 基于 双重检查锁的思想 自己进行了一个创造，也能极大地减少当前服务器的所竞争关系，并且当前服务器上，如果缓存存在，
    //永远不会有锁的问题，如果缓存消失，只有一条线程进行真正的查询。但是，大的商城项目我有万级别的节点个数，synchronized只能锁定
    //当前服务器，管不到其他服务器，其他服务器也得有一条线程查询DB。
    // 如果你的服务器节点数量少，用这种方式，easy。 但是如果是节点个数很多的情况下，这个东西够呛。synchronized是一把本地锁。
    @Deprecated
    private Map<String, CommodityAttributes> getAttrsCacheSync() {
        Object cache = redisCommonProcessor.get("attrsCache");
        if(cache != null) {
            return (Map<String, CommodityAttributes>) cache;
        }
        synchronized (this) {
            if(redisCommonProcessor.get("attrsCache") != null) { //有点双重检查所得意思
                return (Map<String, CommodityAttributes>) redisCommonProcessor.get("attrsCache");
            }
            Map<String, CommodityAttributes> attrsCache = new HashMap<>();
            List<CommodityAttr> attrs = commodityAttrRepository.findAll();
            attrs.forEach(attr -> {
                Optional<CommodityAttrGroup> attrGroup = commodityAttrGroupRepository.findById(attr.getGroupId());
                CommodityAttributes commodityAttributes = CommodityAttributes.builder()
                        .attrName(attr.getName())
                        .attrGroupName(attrGroup.get().getName())
                        .attrGroupId(attrGroup.get().getId())
                        .build();
                attrsCache.put(attr.getId().toString(), commodityAttributes);
            });

            redisCommonProcessor.set("attrsCache", attrsCache);
            return attrsCache;
        }
    }

    //既然synchronized是一把本地锁，那么我们何不考虑用 redis 进行加锁呢？ redis就是一个公共区域的锁了，而且可以做分布式锁。
    // 分布式锁： 就是无论你有多少个服务节点，所有的服务节点都用同一把锁。
     // 虽然还是 Deprecated，不推荐使用。但是 getAttrsCacheRedisSetNx 比 sync 好多了。
    // 如果锁释放出现未知异常，或者释放时间太长，导致所有的相关请求都会无法正常获取结果。
    // 所有的代码和脚本，都是我们自己进行开发的，风险高。
    // 如果删除锁发生了未知异常，我们没有相关的补救措施，如果想进行一些重试机制，还需要继续写代码逻辑。
    // lua脚本和 redis兼容吗？ 2020年写的这个代码。3年过去了，公司要升级 redis，你确定你能想起来吗？
    @Deprecated
    private Map<String, CommodityAttributes> getAttrsCacheRedisSetNx() {
        Object cache = redisCommonProcessor.get("attrsCache");
        if(cache != null) {
            return (Map<String, CommodityAttributes>) cache;
        }
        String uuid = UUID.randomUUID().toString();
        if(redisCommonProcessor.getLock("attrsCacheLock", uuid, 60, TimeUnit.SECONDS)) { //获取锁
            try {
                Map<String, CommodityAttributes> attrsCache = new HashMap<>();
                List<CommodityAttr> attrs = commodityAttrRepository.findAll();
                attrs.forEach(attr -> {
                    Optional<CommodityAttrGroup> attrGroup = commodityAttrGroupRepository.findById(attr.getGroupId());
                    CommodityAttributes commodityAttributes = CommodityAttributes.builder()
                            .attrName(attr.getName())
                            .attrGroupName(attrGroup.get().getName())
                            .attrGroupId(attrGroup.get().getId())
                            .build();
                    attrsCache.put(attr.getId().toString(), commodityAttributes);
                });
                redisCommonProcessor.set("attrsCache", attrsCache);
                return attrsCache;
            } finally {
                //锁的释放. 无论任何情况下都要有锁能够释放出来的逻辑。
                String luaScripts = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                Long res = redisCommonProcessor.runLuaScripts(luaScripts, uuid, "attrsCacheLock");
                System.out.println("del result = " + res);
            }
        } else {
            //要考虑，我们等唯一拿到锁的一条线程执行完成，才可以打破这个递归，而且无限的递归导致JVM stack max local stack异常
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return getAttrsCacheRedisSetNx();
        }
    }

    //redisson 更加成熟的，更加具有封装性的，行业认可的第三方jar包。分布式锁，肯定大概率都会选择 redisson
    private Map<String, CommodityAttributes> getAttrsCacheRedisson() {
        Object cache = redisCommonProcessor.get("attrsCache");
        if(cache != null) {
            return (Map<String, CommodityAttributes>) cache;
        }
        RLock lock = redissonClient.getLock("attrsCacheLock");
//        lock.lock(); // lock 方法不带参数，磨人的超时时间是 30s。Redission每隔10s，给你续期到30s，看门狗模式。一直续期不一定是好事儿
        lock.lock(10, TimeUnit.SECONDS); //尽量保证超时时间 > 业务执行时间。
        //我们可以设置的时间大一些，比如说120s，这段时间肯定能执行完成了。
        // 10s 过期时间，看门狗不会进行续期。
        // 比如说我们的代码执行时间是60s。 那么10s以后，锁自动解锁（再过50s我们的线程执行完了，会调用finally里中的lock.unlock();
        // 会抛出异常：显示当前的锁无法被当前线程解锁。）。 此时另外一条线程过来了，能够拿到锁。出现两条线程同时进行db的查询，
        try {
            if(redisCommonProcessor.get("attrsCache") != null) { //有点双重检查所得意思
                return (Map<String, CommodityAttributes>) redisCommonProcessor.get("attrsCache");
            }
            Map<String, CommodityAttributes> attrsCache = new HashMap<>();
            List<CommodityAttr> attrs = commodityAttrRepository.findAll();
            attrs.forEach(attr -> {
                Optional<CommodityAttrGroup> attrGroup = commodityAttrGroupRepository.findById(attr.getGroupId());
                CommodityAttributes commodityAttributes = CommodityAttributes.builder()
                        .attrName(attr.getName())
                        .attrGroupName(attrGroup.get().getName())
                        .attrGroupId(attrGroup.get().getId())
                        .build();
                attrsCache.put(attr.getId().toString(), commodityAttributes);
            });

            redisCommonProcessor.set("attrsCache", attrsCache);
            return attrsCache;
        } catch (Exception e) {
            throw new UnsupportedOperationException("get attrs cache failed with exception = " + e);
        } finally {
            try {
                lock.unlock();
            } catch (Exception e) {
                System.out.println("打印warning日志即可。");
            }
        }
    }

    @Deprecated //减少与db的交互次数
    public void onSale2(Long spuId) {
        List<Map<String, Object>> esBodies = commoditySpuRepository.getCommodityBySpuId(spuId);
        if(esBodies == null || esBodies.size() == 0) {
            return;
        }
        JSONArray jsonArray = new JSONArray();
        jsonArray.addAll(esBodies);
        List<CommodityEsBodyDeprecated> bodies = jsonArray.toJavaList(CommodityEsBodyDeprecated.class);
        Map<String, CommodityAttributes> attrsCache = getAttrsCacheRedisson();
        List<CommodityAttributes> commonAttrs = new ArrayList<>();
        for (CommodityEsBodyDeprecated commodityEsBodyDeprecated : bodies) {
            // CommodityEsBody commodityEsBody = new CommodityEsBody(); commodityEsBody = JsonUtils.readValue.. 不行
            // clone 方法也不行
            if(commonAttrs.size() == 0) {
                Map<String, String> genericAttrs = JsonUtils.readValue(commodityEsBodyDeprecated.getGenericAttrs(), Map.class);
                genericAttrs.keySet().stream().forEach(key -> {
                    CommodityAttributes commodityAttributes = attrsCache.get(key);
                    commodityAttributes.setAttrId(Long.parseLong(key));
                    commodityAttributes.setAttrValue(genericAttrs.get(key));
                    commonAttrs.add(commodityAttributes);
                });
            }
            List<CommodityAttributes> attrs = new ArrayList<>(commonAttrs);
            Map<String, String> ownAttr = JsonUtils.readValue(commodityEsBodyDeprecated.getOwnAttr(), Map.class);
            ownAttr.keySet().stream().forEach(key -> {
                CommodityAttributes commodityAttributes = attrsCache.get(key);
                attrs.add(CommodityAttributes.builder()
                        .attrGroupId(commodityAttributes.getAttrGroupId()).attrGroupName(commodityAttributes.getAttrGroupName())
                        .attrId(Long.parseLong(key)).attrName(commodityAttributes.getAttrName()).attrValue(ownAttr.get(key)).build());
            });
            commodityEsBodyDeprecated.setAttrs(attrs);
            commodityEsBodyDeprecated.setGenericAttrs(null);
            commodityEsBodyDeprecated.setOwnAttr(null);
        }
        System.out.println(JsonUtils.toJSONString(bodies));
    }

    @Deprecated //冲破 不成文的代码书写规范 的牢笼。
    public List<Map<String, Object>> onSale3(Long spuId) {
        List<Map<String, Object>> esBodies = commoditySpuRepository.getCommodityBySpuId(spuId);
        List<Map<String, Object>> results = new ArrayList<>();
        if(esBodies == null || esBodies.size() == 0) {
            return esBodies;
        }
        Map<String, CommodityAttributes> attrsCache = getAttrsCacheRedisson();
        List<CommodityAttributes> commonAttrs = new ArrayList<>();
        for (Map<String, Object> bodyMap : esBodies) {
            // CommodityEsBody commodityEsBody = new CommodityEsBody(); commodityEsBody = JsonUtils.readValue.. 不行
            // clone 方法也不行
            if(commonAttrs.size() == 0) {
                Map<String, String> genericAttrs = JsonUtils.readValue(bodyMap.get("genericAttrs").toString(), Map.class);
                genericAttrs.keySet().stream().forEach(key -> {
                    CommodityAttributes commodityAttributes = attrsCache.get(key);
                    commodityAttributes.setAttrId(Long.parseLong(key));
                    commodityAttributes.setAttrValue(genericAttrs.get(key));
                    commonAttrs.add(commodityAttributes);
                });
            }
            List<CommodityAttributes> attrs = new ArrayList<>(commonAttrs);
            Map<String, String> ownAttr = JsonUtils.readValue(bodyMap.get("ownAttr").toString(), Map.class);
            ownAttr.keySet().stream().forEach(key -> {
                CommodityAttributes commodityAttributes = attrsCache.get(key);
                attrs.add(CommodityAttributes.builder()
                        .attrGroupId(commodityAttributes.getAttrGroupId()).attrGroupName(commodityAttributes.getAttrGroupName())
                        .attrId(Long.parseLong(key)).attrName(commodityAttributes.getAttrName()).attrValue(ownAttr.get(key)).build());
            });
            Map<String, Object> newMap = new HashMap<>(bodyMap);
            newMap.put("attrs", attrs);
            results.add(newMap);
        }
        return results;
    }

    // 之前 onsale所有的方法，都是采用的类似于 “行存储格式”的内容传递；
    // 此次，我们采用类似于“列式存储格式”的内容传递。
    // 消息压缩--不考虑压缩。
    public CommonResponse onSale4(Long spuId) {
        //1. 查询Db 获取所有的 sku信息
        List<Map<String, Object>> esBodies = commoditySpuRepository.getCommodityBySpuId(spuId);
        if(esBodies == null || esBodies.size() == 0) {
            return ResponseUtils.okResponse(null);
        }
        //2. 组装columns
        Map<String, Object> onSaleBody = new HashMap<>();
        List<Object> columns = generateColumns(esBodies);
        onSaleBody.put("columns", columns);
        //3. 组装 commonAttrs
        List<CommodityAttributes> commonAttrs = generateAttrs(esBodies.get(0), "genericAttrs");
        onSaleBody.put("commonAttrs", commonAttrs);
        //4. 组装 values
        int skuIdIndex = columns.indexOf("skuId");
        final List<Object> skuIds = new ArrayList<>();
        List<Object> values = generateValues(esBodies, columns, skuIds, skuIdIndex);
        onSaleBody.put("values", values);
        //5. 返回值处理及补偿机制
        return responseHandler(onSaleBody, skuIds);
    }

    private CommonResponse responseHandler(Map<String, Object> onSaleBody, List<Object> skuIds) {
        CommonResponse response = productServiceClient.onSaleToEs(onSaleBody);
        if(response.getCode() == ResponseCode.SUCCESS.getCode().intValue()) {
            //更新mysql DB中的上架情况，sku表中的 is_valid
            //返回值处理成完成成功
            try {
                commoditySkuRepository.updateSkuOnSaleStatus(skuIds);
            } catch (Exception e) {
                errorRetryHandler(skuIds);
            }
            return ResponseUtils.okResponse(null);
        } else if (response.getCode() == ResponseCode.BAD_ESBULK_REQUEST.getCode().intValue()) {
            List<Object> diffList = null;
            List<String> failedSkuIds = JsonUtils.readValueFromObject(response.getContent(), List.class);
            try {
                diffList = skuIds.stream()
                        .filter(skuId -> !failedSkuIds.contains(skuId.toString()))
                        .collect(Collectors.toList());
                commoditySkuRepository.updateSkuOnSaleStatus(diffList);
            } catch (Exception e) {
                errorRetryHandler(diffList);
            }
            return ResponseUtils.failResponse(ResponseCode.BAD_ESBULK_REQUEST.getCode(), failedSkuIds,ResponseCode.BAD_ESBULK_REQUEST.getMessage());
        }
        return response;
    }

    private void errorRetryHandler(List<Object> skuIds) {
        List<CommoditySkuUpError> skuUpErrors = new ArrayList<>();
        skuIds.forEach(skuId -> {
            skuUpErrors.add(CommoditySkuUpError.builder().skuId(Integer.parseInt(skuId.toString())).epoch(0).status(0).build());
        });
        commoditySkuUpErrorRepository.saveAll(skuUpErrors);
    }

    private List<Object> generateValues(List<Map<String, Object>> esBodies, List<Object> columns, List<Object> skuIds, int skuIdIndex) {
        List<Object> values = new ArrayList<>();
        esBodies.forEach(body -> {
            List<Object> valueList = new ArrayList<>();
            columns.forEach(col -> {
                valueList.add(body.get(col));
            });
            skuIds.add(valueList.get(skuIdIndex));
            List<CommodityAttributes> attrs = generateAttrs(body, "ownAttr");
            valueList.set(valueList.size()-1, attrs);
            values.add(valueList);
        });
        return values;
    }

    private List<CommodityAttributes> generateAttrs(Map<String, Object> body, String attrKey) {
        Map<String, CommodityAttributes> attrsCache = getAttrsCacheRedisson();
        List<CommodityAttributes> attributes = new ArrayList<>();
        Map<String, String> attrs = JsonUtils.readValue(body.get(attrKey).toString(), Map.class);
        attrs.keySet().stream().forEach(key -> {
            CommodityAttributes commodityAttributes = attrsCache.get(key);
            attributes.add(CommodityAttributes.builder()
                    .attrGroupId(commodityAttributes.getAttrGroupId()).attrGroupName(commodityAttributes.getAttrGroupName())
                    .attrId(Long.parseLong(key)).attrName(commodityAttributes.getAttrName()).attrValue(attrs.get(key)).build());
        });
        return attributes;
    }

    private List<Object> generateColumns(List<Map<String, Object>> esBodies) {
        List<Object> columns = esBodies.get(0).keySet().stream().collect(Collectors.toList());
        columns.remove("ownAttr");
        columns.remove("genericAttrs");
        columns.add("attrs");
        return columns;
    }

}
